#include "stream.hpp"

#include <algorithm>
#include <cstdio>
#include <cstring>
#include <cassert>
#include "log.hpp"

namespace vl
{

#ifndef NDEBUG
#define CHECK() _invariants()
#else
#define CHECK() (void)0
#endif

buffer::buffer()
: _buffer_base(0), _buffer_size(0), _buffer_limit(0), _buffer(0)
{
	
}

buffer::buffer(byte const* p, size_t s)
: _buffer_base(reinterpret_cast<byte*>(malloc(s))), _buffer_size(s)
, _buffer_limit(_buffer_base + s)
, _buffer(_buffer_base)
{
	memcpy(_buffer_base, p, s);
}

buffer::~buffer()
{
	free(_buffer);
}

void buffer::clear()
{
	_buffer_base = _buffer;
	_buffer_size = 0;
	CHECK();
}

void buffer::resize(size_t newSize)
{
retry:
	CHECK();
	
	byte* p = _buffer_base + newSize; // This would be the past-the-end byte of the new buffer
	if(p > _buffer_limit)
	{
		if(_buffer_base != _buffer)
		{
			// We can move the buffer back, do so and we might avoid having to reallocate
			memmove(_buffer, _buffer_base, _buffer_size);
			_buffer_base = _buffer;
			goto retry;
		}
					
		// The minimum size of the virtual buffer we must have
		size_t minCap = (p - _buffer);

		size_t newCap = (_buffer_limit - _buffer) * 2;
		if(newCap < minCap)
			newCap = minCap;
		// The buffer should be at the beginning of the virtual buffer
		assert(_buffer_base == _buffer);
		_buffer = reinterpret_cast<byte*>(realloc(_buffer, newCap));
		_buffer_limit = _buffer + newCap;
		_buffer_base = _buffer;
	}
			
	_buffer_size = newSize;
	
	CHECK();
}

size_t buffer::size() const
{
	CHECK();
	return _buffer_size;
}

bool buffer::empty() const
{
	CHECK();
	return _buffer_size == 0;
}

void buffer::insert(byte const* p, size_t s)
{
	CHECK();
	
	size_t oldSize = size();
	resize(oldSize + s);
	memcpy(_buffer_base + oldSize, p, s);
	
	CHECK();
}

void buffer::insert(byte b, size_t s)
{
	CHECK();
	
	size_t oldSize = size();
	resize(oldSize + s);
	memset(_buffer_base + oldSize, b, s);
	
	CHECK();
}
	
void buffer::consume(size_t amount)
{
	CHECK();
	// We delay the move until it would make
	// us have to enlarge the capacity.
	// That way we have to make much fewer moves
	// of much smaller amounts of data without
	// using more memory.
	
	size_t curSize = size();
	if(amount > curSize)
		throw std::runtime_error("Consumption overflow");
	_buffer_base += amount;
	_buffer_size -= amount;
	CHECK();
}

byte& buffer::operator[](size_t idx)
{
	CHECK();
	return _buffer_base[idx];
}

byte const& buffer::operator[](size_t idx) const
{
	CHECK();
	return _buffer_base[idx];
}

void buffer::_invariants() const
{
	// These conditions always hold:
	// _buffer <= _buffer_base <= (_buffer_base + _buffer_size) <= _buffer_limit

	assert(_buffer <= _buffer_base);
	assert(_buffer_base <= _buffer_base + _buffer_size);
	assert(_buffer_base + _buffer_size <= _buffer_limit);
}

#undef CHECK

/////

byte* device_buf::write_buffer(size_t size)
{
	size_t oldSize = write_buf.size();
	write_buf.resize(oldSize + size);
	return &write_buf[oldSize];
}

// Writes the bytes in [p, p + size) to the stream
void device_buf::write_amount(byte const* p, size_t size)
{
	if(write_buf.empty() && size > write_buffer_size)
	{
		// Try to bypass buffering to avoid unnecessary copying of larger blocks
		size_t written = write(p, size);
		assert(written <= size);
		
		size -= written;
		if(size == 0)
			return; // We wrote all of it successfully, nothing to buffer so return
		p += written;
	}
	
	// Insert into stream
	write_buf.insert(p, size);
	if(write_buf.size() > write_buffer_size)
		flush(); // Time to flush
}

// Tries to write the buffered data to the stream
bool device_buf::flush()
{
	if(write_buf.empty())
		return true;
	
	size_t written = write(&write_buf[0], write_buf.size());

	assert(written <= write_buf.size());
	write_buf.consume(written);
	
	write_buffer_size = std::max(std::size_t(1024ul), write_buf.size() * std::size_t(2ul)); // TODO: What factor should be used here?
	return write_buf.empty();
}

void device_buf::do_close()
{
	
}

void device_buf::close()
{
	flush();
	do_close();
}

/////

device_buf::device_buf()
: write_buffer_size(1024), _reached_eof(false), blocking(0)
{
}

device_buf::device_buf(byte const* p, size_t s)
: read_buf(p, s), _reached_eof(false), blocking(0)
{
}

device_buf::~device_buf()
{
	//close(); // We can't close here because it might make us call a pure-virtual function
}

// Returns the current buffer size
size_t device_buf::read_size() const
{
	return read_buf.size();
}

// Tries to buffer //size// bytes. Returns true if successful, false otherwise.
// This function may block until the request is fulfilled if a blocking policy is specified.
bool device_buf::read_amount(std::size_t size)
{
	if(read_size() >= size)
		return true;

	while(true)
	{
		try_read_amount(size);

		if(read_size() >= size)
			return true;
			
		if(_reached_eof || error()) // TODO: Check for error as well
			return false;
			
		if(blocking)
		{
			flush(); // Perhaps we don't get any data because some request is still in the pipeline, flush
			blocking(this);
			continue;
		}
		return false;
	}
}

bool device_buf::eof() const
{
	return _reached_eof;
}
	
// Tries to buffer //size// bytes. Returns true if any bytes were buffered.
bool device_buf::try_read_amount(std::size_t size)
{
	if(size < read_buf.size())
		return true;

	std::size_t oldSize = read_buf.size();
		
	std::size_t add = size - oldSize;
	
	if(add < read_buffer_size)
	{
		add = read_buffer_size; // We try to read at least buffer_size bytes
		size = oldSize + add;
	}

	read_buf.resize(size);
	
	ASSERT(size == read_buf.size());
	ASSERT(size == oldSize + add);
	std::size_t amountRead = read(&read_buf[oldSize], add);
	//assert(amountRead <= size);

	if(amountRead == device_buf::eof_value)
	{
		DLOG("Reached EOF");
		read_buf.resize(oldSize);
		_reached_eof = true;
		return false;
	}
	read_buf.resize(oldSize + amountRead);
	return amountRead > 0;
}

void device_buf::read_pad(size_t size)
{
	read_buf.insert(byte(0), size);
}

void device_buf::read_direct_fill(byte const* p, size_t size)
{
	read_buf.insert(p, size);
}

// Returns a pointer to a buffer read_size() long
byte const* device_buf::read_buffer() const
{
	return &read_buf[0];
}

// Consumes the first //size// bytes in the buffer
void device_buf::read_consume(size_t size)
{
	read_buf.consume(size);
}


/////

memory_buf::memory_buf()
{
}

memory_buf::memory_buf(byte const* p, size_t s)
: device_buf(p, s)
{
}


byte* memory_buf::write_buffer(size_t size)
{
	size_t oldSize = read_buf.size();
	read_buf.resize(oldSize + size);
	return &read_buf[oldSize];
}

// Writes the bytes in [p, p + size) to the stream
void memory_buf::write_amount(byte const* p, size_t size)
{
	read_buf.insert(p, size);
}

size_t memory_buf::read(byte* p, size_t size)
{
	return device_buf::eof_value;
}

size_t memory_buf::write(byte const* p, size_t size)
{
	return size;
}

/////

bool extract_line(device_buf* src, std::string& str)
{
	if(src->eof())
		return false;
	
	std::size_t limit = src->read_size();
	byte const* buf = src->read_buffer();
	for(size_t i = 0; ; ++i)
	{
		if(i >= limit)
		{
			if(!src->try_read_amount(src->read_size() + 256))
			{
				// No more bytes available
				if(src->eof())
				{
					// eof, return whole buffer
					if(src->read_size() > 0)
					{
						str.assign((char*)src->read_buffer(), src->read_size());
						src->read_consume(src->read_size());
						return true;
					}
				}
				return false;
			}
			limit = src->read_size();
			buf = src->read_buffer();
		}
		
		char b = buf[i];
		if(b == '\n' || b == '\r')
		{
			if(i != 0)
			{
				str.assign((char*)src->read_buffer(), i);
				src->read_consume(i + 1); // Consume new-line too
				return true;
			}
			else
			{
				// Ignore zero-length lines
				src->read_consume(1);
				i = 0;
				limit = src->read_size();
				buf = src->read_buffer();
			}
		}
	}
}

}

